package com.sheepone.distributor.config;

import com.lmax.disruptor.EventTranslator;
import com.sheepone.distributor.DefaultDistributor;
import com.sheepone.distributor.DelayDistributor;
import com.sheepone.distributor.Distributor;
import com.sheepone.distributor.DistributorLine;
import com.sheepone.distributor.event.DistributeEvent;
import com.sheepone.distributor.expand.DataMonitor;
import com.sheepone.distributor.listener.DistributorListener;
import com.sheepone.distributor.thread.DelayEventQueue;
import com.sheepone.util.algorithm.analysis.IntegerAnalysisUtil;
import org.apache.commons.lang.ClassUtils;

import java.util.ArrayList;

/**
 * @author Miss.杨
 * @description
 * @since 2024/2/15 - 16:49
 */
public class DistributorCreator {
    public static Distributor create(DistributorLineConfig distributorLineConfig) {
        int num = distributorLineConfig.getHandlerNum();// DistributorLine的数量
        ArrayList<DistributorLine> distributorLines = new ArrayList<>(num);

        // 均分后每个DistributorLine的最小的容量，应该为2的整次幂
        int lineConfigSize = distributorLineConfig.getSize();
        String name = distributorLineConfig.getName();
        if (lineConfigSize < 1) {
            throw new IllegalArgumentException("The size of the event line \"" + name + "\" should greater (>) than 0");
        }
        int size = IntegerAnalysisUtil.upper2Pow(lineConfigSize / num);
        DelayEventQueue<EventTranslator<DistributeEvent>> delayEventQueue = new DelayEventQueue<>();
        for (int i = 0; i < num; i++) {
            distributorLines.add(new DistributorLine(name, size, delayEventQueue));
        }

        Distributor distributor;
        if (distributorLineConfig.isEnableDelay()) {
            distributor = new DelayDistributor(name, lineConfigSize, distributorLines);
        } else {
            distributor = new DefaultDistributor(name, lineConfigSize, distributorLines);
        }
        handlerDistributorListener(distributorLineConfig, distributor);

        return distributor;
    }

    private static void handlerDistributorListener(DistributorLineConfig distributorLineConfig, Distributor distributor) {
        int intervalTime = distributorLineConfig.getIntervalTime();

        if (DistributorLineConfig.enableDataMonitor(intervalTime)) {
            distributor.addListeners(new DataMonitor(intervalTime));
        }
        DistributorListener[] distributorListeners = findDistributorListeners(distributorLineConfig.getListeners());
        if (distributorListeners != null) {
            distributor.addListeners(distributorListeners);
        }
    }

    private static DistributorListener[] findDistributorListeners(String[] listeners) {
        if (listeners == null || listeners.length < 1) {
            return null;
        }
        ArrayList<DistributorListener> objects = new ArrayList<>();
        for (String s : listeners) {
            try {
                Class<?> listenerClass = Class.forName(s);
                // 如果父类和子类的classloader不一样，Class.isAssignableFrom方法就会失效
                if (ClassUtils.isAssignable(listenerClass, DistributorListener.class)) {
                    objects.add((DistributorListener) listenerClass.getConstructor().newInstance());
                }
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
        return objects.toArray(value -> new DistributorListener[0]);
    }
}